package rocket

import (
	"fmt"
	mq_http_sdk "github.com/aliyunmq/mq-http-go-sdk"
	"github.com/gogap/errors"
	"github.com/zeromicro/go-zero/core/logx"
	"github.com/zeromicro/go-zero/core/queue"
	"github.com/zeromicro/go-zero/core/service"
	"github.com/zeromicro/go-zero/core/stat"
	"github.com/zeromicro/go-zero/core/threading"
	"github.com/zeromicro/go-zero/core/timex"
	"log"
	"strings"
	"time"
)

type (
	RqConf struct {
		service.ServiceConf
		InstanceId string
		Topic      string
		GroupId    string
		Tag        string
		Conns      int
	}

	ConsumerConf struct {
		service.ServiceConf
		InstanceId string
		Topic      string
		GroupId    string
		Tag        string
		Client     RocketClient
		Conns      int
	}

	ConsumeHandle func(key, body string) error

	ConsumeHandler interface {
		Consume(key, body string) error
	}

	rocketOptions struct {
		waitSecond  int64
		numberOfMsg int32
		consumers   int
		metrics     *stat.Metrics
	}
	RocketOption func(options *rocketOptions)

	rocketQueue struct {
		c                ConsumerConf
		consumer         mq_http_sdk.MQConsumer
		handler          ConsumeHandler
		producerRoutines *threading.RoutineGroup
		consumerRoutines *threading.RoutineGroup
		respChannel      chan mq_http_sdk.ConsumeMessageResponse
		endChannel       chan int
		errChan          chan error
		numberOfMsg      int32
		waitSeconds      int64
		consumers        int
		metrics          *stat.Metrics
	}

	rocketQueues struct {
		queues []queue.MessageQueue
		group  *service.ServiceGroup
	}
)

func MustNewQueue(c ConsumerConf, handler ConsumeHandler, opts ...RocketOption) queue.MessageQueue {
	q, err := NewQueue(c, handler, opts...)
	if err != nil {
		log.Fatal(err)
	}
	return q
}
func NewQueue(c ConsumerConf, handler ConsumeHandler, opts ...RocketOption) (queue.MessageQueue, error) {
	if err := c.SetUp(); err != nil {
		return nil, err
	}

	// Options
	var options rocketOptions
	for _, opt := range opts {
		opt(&options)
	}
	ensureRocketOptions(c, &options)

	if c.Conns < 1 {
		c.Conns = 1
	}
	q := rocketQueues{
		group: service.NewServiceGroup(),
	}
	for i := 0; i < c.Conns; i++ {
		q.queues = append(q.queues, newRocketQueue(c, handler, options))
	}

	return q, nil
}
func newRocketQueue(c ConsumerConf, handler ConsumeHandler, options rocketOptions) queue.MessageQueue {
	consumer := c.Client.MQClient.GetConsumer(c.InstanceId, c.Topic, c.GroupId, c.Tag)
	return &rocketQueue{
		c:                c,
		consumer:         consumer,
		handler:          handler,
		producerRoutines: threading.NewRoutineGroup(),
		consumerRoutines: threading.NewRoutineGroup(),
		respChannel:      make(chan mq_http_sdk.ConsumeMessageResponse),
		//endChannel:       make(chan int),
		errChan:     make(chan error),
		numberOfMsg: options.numberOfMsg,
		waitSeconds: options.waitSecond,
		consumers:   options.consumers,
		metrics:     options.metrics,
	}
}
func (q *rocketQueue) Start() {

	q.subscribeConsumers()

	//q.startConsumers()
	//q.startProducers()

	//q.producerRoutines.Wait()
	//close(q.respChannel)
	//close(q.errChan)
	//close(q.endChannel)
	//q.consumerRoutines.Wait()
}
func (q *rocketQueue) Stop() {

}
func (q *rocketQueue) consumeOne(key, val string) error {
	startTime := timex.Now()
	err := q.handler.Consume(key, val)
	q.metrics.Add(stat.Task{
		Duration: timex.Since(startTime),
	})
	return err
}

func (q *rocketQueue) subscribeConsumers() {
	for i := 0; i < q.consumers; i++ {
		go func() {
			for {
				endChan := make(chan int)
				respChan := make(chan mq_http_sdk.ConsumeMessageResponse)
				errChan := make(chan error)
				go func() {
					select {
					case resp := <-respChan:
						{
							fmt.Println("check mqs message =======> ")

							//for msg := range q.respChannel {
							fmt.Println("receive mqs message =====> ")
							fmt.Println(resp.Messages)
							// 处理业务逻辑。
							var handles []string
							for _, v := range resp.Messages {
								if err := q.consumeOne(v.MessageKey, v.MessageBody); err != nil {
									logx.Errorf("Error on consuming: %s, error: %v", v.MessageBody, err)
								} else {
									handles = append(handles, v.ReceiptHandle)
								}

							}
							q.mqConsumerAckMsg(handles)
							endChan <- 1
							//q.endChannel <- 1
							//}

						}
					case err := <-errChan:
						{
							// Topic中没有消息可消费。
							if strings.Contains(err.(errors.ErrCode).Error(), "MessageNotExist") {
								//fmt.Println("\nNo new message, continue! - tag - ", q.consumer.MessageTag())
							} else {
								logx.Error(err)
								time.Sleep(time.Duration(3) * time.Second)
							}
							endChan <- 1
						}
					case <-time.After(35 * time.Second):
						{
							logx.Error("Timeout of consumer message ??")
							endChan <- 1
						}
					}
				}()
				// 长轮询消费消息，网络超时时间默认为35s。
				// 长轮询表示如果Topic没有消息，则客户端请求会在服务端挂起3s，3s内如果有消息可以消费则立即返回响应。
				q.consumer.ConsumeMessage(respChan, errChan,
					q.numberOfMsg, // 一次最多消费3条（最多可设置为16条）。
					q.waitSeconds, // 长轮询时间3s（最多可设置为30s）。
				)
				<-endChan
			}
		}()
	}
}
func (q *rocketQueue) startConsumers() {
	for i := 0; i < q.consumers; i++ {
		/*
			go func() {
				for {
					fmt.Println("check 123 ====")
					endChan := make(chan int)
					go func() {
						select {
						case resp := <-q.respChannel:
							{
								fmt.Println("check mqs message =======> ")

								//for msg := range q.respChannel {
								fmt.Println("receive mqs message =====> ")
								fmt.Println(resp.Messages)
								// 处理业务逻辑。
								var handles []string
								for _, v := range resp.Messages {
									if err := q.consumeOne(v.MessageKey, v.MessageBody); err != nil {
										logx.Errorf("Error on consuming: %s, error: %v", v.MessageBody, err)
									} else {
										handles = append(handles, v.ReceiptHandle)
									}

								}
								q.mqConsumerAckMsg(handles)
								endChan <- 1
								//q.endChannel <- 1
								//}

							}
						case err := <-q.errChan:
							{
								// Topic中没有消息可消费。
								if strings.Contains(err.(errors.ErrCode).Error(), "MessageNotExist") {
									fmt.Println("\nNo new message, continue!")
								} else {
									logx.Error(err)
									time.Sleep(time.Duration(3) * time.Second)
								}
								endChan <- 1
							}
						case <-time.After(35 * time.Second):
							{
								logx.Error("Timeout of consumer message ??")
								endChan <- 1
							}
						}
					}()
					fmt.Println("wait end chan")
					<-endChan
					q.endChannel <- 1
					fmt.Println("run end chan")
					//time.Sleep(time.Millisecond * 300)
				}
			}()
		*/
		/*
			q.consumerRoutines.Run(func() {
				for true {
					fmt.Println("check mqs message =======> ")
					for msg := range q.respChannel {
						fmt.Println("receive mqs message =====> ")
						fmt.Println(msg.Messages)
						// 处理业务逻辑。
						var handles []string
						for _, v := range msg.Messages {
							if err := q.consumeOne(v.MessageKey, v.MessageBody); err != nil {
								logx.Errorf("Error on consuming: %s, error: %v", v.MessageBody, err)
							} else {
								handles = append(handles, v.ReceiptHandle)
							}

						}
						q.mqConsumerAckMsg(handles)
						//q.endChannel <- 1
					}
				}
			})
		*/

		q.consumerRoutines.Run(func() {
			for msg := range q.respChannel {
				// 处理业务逻辑。
				var handles []string
				for _, v := range msg.Messages {
					if err := q.consumeOne(v.MessageKey, v.MessageBody); err != nil {
						logx.Errorf("Error on consuming: %s, error: %v", v.MessageBody, err)
					} else {
						handles = append(handles, v.ReceiptHandle)
					}

				}
				q.mqConsumerAckMsg(handles)
				q.endChannel <- 1
			}
		})
	}
}

/*
	func (q *rocketQueue) startConsumers() {
		for i := 0; i < q.consumers; i++ {
			q.consumerRoutines.Run(func() {
				for {
					select {
					case msg := <-q.respChannel:
						// 这里需要考虑如何异步处理 并 ack 句柄
						// ack 需要所有的执行结束才能提交 不然会产生 竞争
						//for msg := range q.respChannel {
						// 处理业务逻辑。
						var handles []string
						for _, v := range msg.Messages {
							if err := q.consumeOne(v.MessageKey, v.MessageBody); err != nil {
								logx.Errorf("Error on consuming: %s, error: %v", v.MessageBody, err)
							} else {
								handles = append(handles, v.ReceiptHandle)
							}

						}
						q.mqConsumerAckMsg(handles)
						q.endChannel <- 1
						//}
					case err := <-q.errChan:
						{
							// Topic中没有消息可消费。
							if strings.Contains(err.(errors.ErrCode).Error(), "MessageNotExist") {
								fmt.Println("\nNo new message, continue!")
							} else {
								logx.Error(err)
								time.Sleep(time.Duration(3) * time.Second)
							}
							q.endChannel <- 1
						}
					case <-time.After(35 * time.Second):
						{
							logx.Error("Timeout of consumer message ??")
							q.endChannel <- 1
						}
					}
				}

			})
		}

}
*/
func (q *rocketQueue) mqConsumerAckMsg(handles []string) {
	// NextConsumeTime前若不确认消息消费成功，则消息会被重复消费。
	// 消息句柄有时间戳，同一条消息每次消费拿到的都不一样。
	ackerr := q.consumer.AckMessage(handles)
	if ackerr != nil {
		// 某些消息的句柄可能超时，会导致消息消费状态确认不成功。
		//fmt.Println(ackerr)
		logx.Error(ackerr)
		if errAckItems, ok := ackerr.(errors.ErrCode).Context()["Detail"].([]mq_http_sdk.ErrAckItem); ok {
			for _, errAckItem := range errAckItems {
				logx.Errorf("\tErrorHandle:%s, ErrorCode:%s, ErrorMsg:%s\n",
					errAckItem.ErrorHandle, errAckItem.ErrorCode, errAckItem.ErrorMsg)
			}
		} else {
			logx.Error("ack err =", ackerr)
		}
		time.Sleep(time.Duration(3) * time.Second)
	} else {
		logx.Infof("Ack ---->\n\t%s\n", handles)
	}
}

const (
	defaultWaitSeconds = 3
	defaultNumberOfMsg = 1
)

func (q *rocketQueue) startProducers() {
	go func() {
		for {
			fmt.Println("fetch mqs message =====> ", time.Now())
			q.consumer.ConsumeMessage(q.respChannel, q.errChan, q.numberOfMsg, q.waitSeconds)
			<-q.endChannel

			fmt.Println("fetch mqs message =====> end")
		}
	}()
	//q.producerRoutines.Run(func() {
	//	for {
	//		fmt.Println("fetch mqs message =====> ", time.Now())
	//		q.consumer.ConsumeMessage(q.respChannel, q.errChan, q.numberOfMsg, q.waitSeconds)
	//		<-q.endChannel
	//		fmt.Println("fetch mqs message =====> end")
	//	}
	//})
	//for {
	//	q.consumer.ConsumeMessage(q.respChannel, q.errChan, q.numberOfMsg, q.waitSeconds)
	//	//time.Sleep(time.Second)
	//}

}

// Option
func WithWaitSecond(wait int64) RocketOption {
	return func(options *rocketOptions) {
		options.waitSecond = wait
	}
}

func WithNumberOfMessage(num int) RocketOption {
	return func(options *rocketOptions) {
		options.numberOfMsg = int32(num)
	}
}

func WithConsumerCount(count int) RocketOption {
	return func(options *rocketOptions) {
		options.consumers = count
	}
}

func WithMetrics(metrics *stat.Metrics) RocketOption {
	return func(options *rocketOptions) {
		options.metrics = metrics
	}
}

func (q rocketQueues) Start() {
	for _, each := range q.queues {
		q.group.Add(each)
	}
	q.group.Start()
}

func (q rocketQueues) Stop() {
	q.group.Stop()
}

// default filter
func ensureRocketOptions(c ConsumerConf, options *rocketOptions) {
	if options.waitSecond == 0 {
		options.waitSecond = defaultWaitSeconds
	}
	if options.numberOfMsg == 0 {
		options.numberOfMsg = defaultNumberOfMsg
	}
	if options.consumers == 0 {
		options.consumers = defaultNumberOfMsg
	}
	if options.metrics == nil {
		options.metrics = stat.NewMetrics(c.Name)
	}
}
